package cn.doitedu.api;


import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/5
 * @Desc: 学大数据，上多易教育
 *
 *   从kafka中读取如下数据：
 *     {"play_id":"001","user_id":1,"event_id":"video_play","action_time":1000}
 *     {"play_id":"001","user_id":1,"event_id":"video_heartbeat","action_time":6000}
 *     {"play_id":"001","user_id":1,"event_id":"video_heartbeat","action_time":11000}
 *     {"play_id":"001","user_id":1,"event_id":"video_heartbeat","action_time":16000}
 *     {"play_id":"001","user_id":1,"event_id":"video_pause","action_time":18000}
 *
 *     {"play_id":"001","user_id":1,"event_id":"video_resume","action_time":30000}
 *     {"play_id":"001","user_id":1,"event_id":"video_heartbeat","action_time":35000}
 *     {"play_id":"001","user_id":1,"event_id":"video_heartbeat","action_time":40000}
 *     {"play_id":"001","user_id":1,"event_id":"video_heartbeat","action_time":45000}
 *     {"play_id":"001","user_id":1,"event_id":"video_pause","action_time":47000}
 *
 *     {"play_id":"001","user_id":1,"event_id":"video_resume","action_time":60000}
 *     {"play_id":"001","user_id":1,"event_id":"video_heartbeat","action_time":65000}
 *     {"play_id":"001","user_id":1,"event_id":"video_heartbeat","action_time":70000}
 *     {"play_id":"001","user_id":1,"event_id":"video_stop","action_time":72000}
 *
 *   计算每个用户，每次视频播放的有效时长(不包含暂停的那一段时长）；
 *
 *
 **/
public class _36_FlinkSql_Graduate_Exercise {


    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        DataStreamSource<String> stream = env.socketTextStream("localhost", 8899);
        SingleOutputStreamOperator<VideoPlay> beans = stream.map(json -> JSON.parseObject(json, VideoPlay.class));

        tenv.createTemporaryView("events",beans,
                Schema.newBuilder()
                        .column("play_id", DataTypes.STRING())
                        .column("user_id", DataTypes.INT())
                        .column("event_id", DataTypes.STRING())
                        .column("action_time", DataTypes.BIGINT())
                        .columnByExpression("rt","to_timestamp_ltz(action_time,3)")
                        .watermark("rt","rt")
                        .build());


        tenv.executeSql(
                "create temporary view abc as  select\n" +
                "     user_id,\n" +
                "     play_id,\n" +
                "     max(action_time)-min(action_time) as timelong\n" +
                "from (\n" +
                "     select\n" +
                "         play_id,\n" +
                "         user_id,\n" +
                "         event_id,\n" +
                "         action_time,\n" +
                "         sum(flag) over(partition by play_id,user_id order by rt ) as flag\n" +
                "     from (\n" +
                "         select\n" +
                "             play_id,\n" +
                "             user_id,\n" +
                "             event_id,\n" +
                "             action_time,\n" +
                "             rt,\n" +
                "             if(event_id = 'video_resume',1,0) as flag\n" +
                "         from events ) o1\n" +
                " ) o2\n" +
                "group by user_id,play_id,flag");


        tenv.executeSql("select user_id,play_id,sum(timelong) as amt_timelong from abc group by user_id,play_id").print();


        env.execute();


    }


    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class VideoPlay{
        private String play_id;
        private int user_id;
        private String event_id;
        private long action_time;
    }

}


